package com.lagou.edu.zookeeper;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.Lists;
import com.lagou.edu.pojo.ServiceNodeInfo;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;

import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.List;

/**
 * @author 赵虎
 * @date 2020/11/29 6:58 下午
 */
public class ZookeeperFactory {

    private static ZookeeperFactory zookeeperFactory;

    private static final String zkConnect = "10.10.10.205:2181";

    private static final String nameSpace = "lg-api";

    public static final String addressSubMark = ":";

    private String servicePath;

    public static final String urlSubMark = "/";

    private CuratorFramework client;

    public String getServicePath() {
        return servicePath;
    }

    public static ZookeeperFactory getInstance(String ip, Integer port) {
        if (zookeeperFactory == null) {
            zookeeperFactory = new ZookeeperFactory(ip, port);
        }
        return zookeeperFactory;
    }

    public static ZookeeperFactory getInstance() {
        if (zookeeperFactory == null) {
            zookeeperFactory = new ZookeeperFactory();
        }
        return zookeeperFactory;
    }

    public ZookeeperFactory() {
        initClient();
    }

    public ZookeeperFactory(String ip, int port) {
        servicePath = ip + addressSubMark + port;
        initClient();
    }


    /**
     * 初始化Zookeeper客户端
     */
    private void initClient() {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        client = CuratorFrameworkFactory.builder()
                .connectString(zkConnect)
                .sessionTimeoutMs(50000)
                .connectionTimeoutMs(30000)
                .retryPolicy(retryPolicy)
                .namespace(nameSpace)
                .build();
        client.start();
    }

    /**
     * 注册服务(自定义服务路径)
     */
    public void registrationService(String path, Object object, CreateMode createMode) throws Exception {
        client.create()
                .creatingParentContainersIfNeeded()
                .withMode(createMode)
                .forPath(path, JSON.toJSONString(object).getBytes(StandardCharsets.UTF_8));
    }

    /**
     * 注册服务(服务列表)
     */
    public void registrationService(List<String> serviceList, CreateMode createMode, String ip, Integer port) throws Exception {
        if (serviceList == null || serviceList.size() == 0) {
            return;
        }
        for (String service : serviceList) {
            String servicePath = getServicePath(service);
            ServiceNodeInfo serviceNodeInfo = new ServiceNodeInfo();
            serviceNodeInfo.setIp(ip);
            serviceNodeInfo.setPort(port);
            serviceNodeInfo.setPath(servicePath);
            serviceNodeInfo.setRespDate(new Date());
            serviceNodeInfo.setUseTime(0L);
            registrationService(servicePath, serviceNodeInfo, createMode);
        }
    }

    /**
     * 获得服务路径
     */
    private String getServicePath(String service) {
        return StringUtils.join(urlSubMark, service, urlSubMark, servicePath);
    }

    /**
     * 删除服务
     */
    public void deleteService(String service) throws Exception {
        client.delete().deletingChildrenIfNeeded().forPath(service);
    }

    /**
     * 删除服务
     */
    public void deleteService(List<String> serviceList) throws Exception {
        for (String service : serviceList) {
            String servicePath = getServicePath(service);
            deleteService(servicePath);
        }
    }

    /**
     * 获取路径值
     */
    public <T> T getServiceValue(String path, Class<T> cls) throws Exception {
        byte[] bytes = client.getData().forPath(path);
        String valueString = new String(bytes);
        return JSON.parseObject(valueString, cls);
    }

    /**
     * 获取路径值
     */
    public Boolean checkExists(String path) throws Exception {
        return client.checkExists().forPath(path) != null;
    }

    /**
     * 获取路径子节点路径
     */
    public List<String> getChildNodePath(String path) throws Exception {
        return client.getChildren().forPath(StringUtils.join(urlSubMark, path));
    }

    /**
     * 获取路径子节点值
     */
    public List<ServiceNodeInfo> getChildNodeValue(String servicePath) throws Exception {
        List<String> childNodeValue = getChildNodePath(servicePath);

        List<ServiceNodeInfo> serviceNodeInfoList = Lists.newArrayList();
        if (childNodeValue != null && childNodeValue.size() > 0) {
            for (String childPath : childNodeValue) {
                String path = StringUtils.join(urlSubMark, servicePath, urlSubMark, childPath);
                serviceNodeInfoList.add(getServiceValue(path, ServiceNodeInfo.class));
            }
        }
        return serviceNodeInfoList;
    }

    /**
     * 更新节点数据
     */
    public void updateServiceValue(String path, Object obj) throws Exception {
        client.setData().forPath(path, JSON.toJSONString(obj).getBytes());
    }

    public String getChildPath(String servicePath, String childrenNode) {
        return StringUtils.join(urlSubMark, servicePath, urlSubMark, childrenNode);
    }

    /**
     * 获得服务节点的子节点监听对象
     */
    public PathChildrenCache getPathChildrenCache(String servicePath) throws Exception {
        PathChildrenCache cache = new PathChildrenCache(client, StringUtils.join(urlSubMark, servicePath), true);
        cache.start();
        return cache;
    }

    /**
     * 获得节点监听对象
     */
    public NodeCache getNodeCache(String path) throws Exception {
        NodeCache nodeCache = new NodeCache(client, path, false);
        nodeCache.start();
        return nodeCache;
    }
}
